package mix

import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.scala._

/**
  * @author dinghao 
  * @create 2021-09-02 13:05 
  * @message
  */
object NcDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val data:DataStream[String] = env.socketTextStream(args(0),9999)
    val value = data.flatMap(_.split(" ")).map((_,1)).filter(_._1 != "nc").keyBy(_._1).reduce((s1,s2)=>(s1._1,s1._2+s2._2))
    new Thread(
      new Runnable {
        override def run(): Unit = {
          value.writeAsText(args(1),FileSystem.WriteMode.OVERWRITE)
        }
      }
    ).start()
    value.print()
    env.execute("NcDemo")
  }
}
